package com.zjl.学习.第07章_高级.B_并行步骤;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.zjl.springbatch.util.OutColour;
import com.zjl.学习.第07章_高级.User;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.job.builder.FlowBuilder;
import org.springframework.batch.core.job.flow.Flow;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.batch.item.json.*;
import org.springframework.batch.item.json.builder.JsonFileItemWriterBuilder;
import org.springframework.batch.item.json.builder.JsonItemReaderBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.PathResource;
import org.springframework.core.task.SimpleAsyncTaskExecutor;

import java.nio.charset.Charset;
import java.util.List;

/**
 * 1：jsonItemReader() flatItemReader() 定义2个读入操作，分别读json格式跟普通文本格式
 *
 * 2：parallelJob() 配置job，需要指定并行的flow步骤，先是parallelFlow1然后是parallelFlow2 ，
 *    2个步骤间使用**.split(new SimpleAsyncTaskExecutor())** 隔开，
 *    表示线程池开启2个线程，分别处理parallelFlow1， parallelFlow2 2个步骤。
 *
 */
@SpringBootApplication
@EnableBatchProcessing
public class B_并行步骤 {
    @Autowired//@EnableBatchProcessing 这个注解帮你从 spring batch 的配置文件中加载了
    private JobLauncher jobLauncher;//Job 启动器
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Bean
    public FlatFileItemReader<User> itemReader_平面读(){
        return new FlatFileItemReaderBuilder<User>()
                .name("itemReaderC")
                .resource(new ClassPathResource("item/user-thread.txt"))  //指定读取的文件
                .delimited().delimiter("#")  //读取出一行数据，该如何分割数据，默认以,分割，当前使用#号分割
                .names("id", "name", "age")//给分割后数据打name标记，后续跟User对象属性进行映射
                .targetType(User.class)      //读取出一行数据封装成什么对象
                .build();
    }
    @Bean
    public JsonItemReader<User> ItemReaderJson读取(){
        ObjectMapper objectMapper = new ObjectMapper();
        JacksonJsonObjectReader<User> jsonObjectReader = new JacksonJsonObjectReader<>(User.class);
        jsonObjectReader.setMapper(objectMapper);

        JsonItemReader<User> userItemReaderJson读取 = new JsonItemReaderBuilder<User>()
                .name("userItemReaderJson读取")
                .jsonObjectReader(jsonObjectReader)
                .resource(new ClassPathResource("item/user.json"))
                .build();
        return userItemReaderJson读取;
    }

    @Bean
    public FlatFileItemWriter<User> itemWriter写入平面文件(){

        return new FlatFileItemWriterBuilder<User>()
                .name("itemWriter写入平面文件")
                .saveState(false)//防止状态被覆盖 Spring Batch 提供大部分的ItemReader是有状态的，作业重启基本通过状态来确定作业停止位置
                //而在多线程环境中，如果对象维护状态被多个线程访问，可能存在线程间状态相互覆盖问题。
                // 所以设置为false表示关闭状态，但这也意味着作业不能重启了
                .resource(new PathResource("D:\\work\\gitee\\spring-batch\\file\\batch\\并行步骤-多线程.txt"))  //输出全路径文件
                .formatted()  //数据格式指定
                .format("id: %s,姓名：%s,年龄：%s")  //输出数据格式
                .names("id", "name", "age")  //需要输出属性
//                .append(true)  //如果输出文件已经存在， 不删除，直接追加到现有文件中
                .build();
    }

    @Bean//输出成  Json 的方法  明确指定Json格式装配器
    public JacksonJsonObjectMarshaller<User> JacksonToJson(){
        JacksonJsonObjectMarshaller marshaller = new JacksonJsonObjectMarshaller();
        return marshaller;
    }

    @Bean//输出成json
    public JsonFileItemWriter<User> itemWriterToJson(){
        return new JsonFileItemWriterBuilder<User>()
                .name("jsonUserItemWriter")
                //防止状态被覆盖 Spring Batch 提供大部分的ItemReader是有状态的，作业重启基本通过状态来确定作业停止位置，
                // 而在多线程环境中，如果对象维护状态被多个线程访问，可能存在线程间状态相互覆盖问题。
                // 所以设置为false表示关闭状态，但这也意味着作业不能重启了
                .saveState(false)
                .resource(new PathResource("D:\\work\\gitee\\spring-batch\\file\\batch\\并行步骤-多线程.json"))
                .jsonObjectMarshaller(JacksonToJson())//使用哪种输出器
//                .append(true)  //如果输出文件已经存在， 不删除，直接追加到现有文件中
                .build();
    }

    @Bean
    public ItemWriter<User> itemWriter(){
        return new ItemWriter<User>() {
            @Override
            public void write(List<? extends User> items) throws Exception {
                OutColour.out.printlnRed(items + "个数：" + items.size());
            }
        };
    }

    @Bean
    public Step stepItem07B平面读() throws Exception {
        return stepBuilderFactory.get("stepItem07B平面读")
                .<User, User>chunk(1)//执行多少次读处理，再进行写
                .reader(itemReader_平面读())
                .writer(itemWriterToJson())//写入json+
                .writer(itemWriter写入平面文件())//写入平面文件
                .writer(itemWriter())//打印
//                .taskExecutor(new SimpleAsyncTaskExecutor())//开启异步多线程读写，无序，但是如json出现争抢问题，导致输出文件格式错误
//                .taskExecutor(new SyncTaskExecutor())//开启同步多线程读写，有顺序适合输出的文件，和线程不安全的输出方式，如写文件
                .build()
                ;
    }
    @Bean
    public Step stepItem07BJson读取() throws Exception {
        return stepBuilderFactory.get("stepItem07BJson读取")
                .<User, User>chunk(1)//执行多少次读处理，再进行写
                .reader(ItemReaderJson读取())
                .writer(itemWriterToJson())//写入json
                .writer(itemWriter())//打印
//                .writer(itemWriter写入平面文件())//写入平面文件
//                .taskExecutor(new SimpleAsyncTaskExecutor())//开启异步多线程读写，无序，但是如json出现争抢问题，导致输出文件格式错误
//                .taskExecutor(new SyncTaskExecutor())//开启同步多线程读写，有顺序适合输出的文件，和线程不安全的输出方式，如写文件
                .build()
                ;
    }
    @Bean
    public Job jobItem07并行() throws Exception {

        //线程1-读user-thread.txt
        Flow parallelFlow1 = new FlowBuilder<Flow>("Flow平面读")
                .start(stepItem07B平面读())
                .build();

        //线程2-读user.json
        Flow parallelFlow2 = new FlowBuilder<Flow>("FlowJson读取")
                .start(stepItem07BJson读取())
                .split(new SimpleAsyncTaskExecutor())//开启
                .add(parallelFlow1)//步骤1和步骤 一起启动
                .build();


        return jobBuilderFactory.get("jobItem07并行1")
                .start(parallelFlow2)
                .end()
                .incrementer(new RunIdIncrementer())
                .build();
    }
    public static void main(String[] args) {
        System.setProperty("spring.profiles.active","01");//  同理 --spring.profiles.active=xxx
        SpringApplication.run(B_并行步骤.class, args);
        System.out.println("***********启动完毕**当前编码:" + Charset.defaultCharset().name() + "****************");

    }
}
